package demo.kafka.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.function.Supplier;

@Configuration
public class ProducerCfg {

    private int messageCount = 0;

    /**
     * 消息供应商 - 定时生成消息
     * 这是一个Supplier函数，会自动向绑定的topic发送消息
     */
    @Bean
    public Supplier<MessageEvent> messageSupplier() {
        return () -> {
            // 每5秒生成一条消息（实际中可以使用Scheduled注解控制频率）
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }

            messageCount++;
            String[] sampleMessages = {
                    "Hello Kafka Streams with Spring Cloud Stream",
                    "This is a test message for stream processing",
                    "Real-time data processing with Kafka",
                    "Spring Cloud Stream makes it easy to work with Kafka",
                    "Building scalable microservices with event streaming"
            };

            String message = sampleMessages[messageCount % sampleMessages.length];

            MessageEvent event = new MessageEvent(
                    UUID.randomUUID().toString(),
                    message,
                    "INFO",
                    LocalDateTime.now(),
                    messageCount
            );

            System.out.println("📤 Produced message: " + event.getMessageId() + " - " + event.getContent());
            return event;
        };
    }

    /**
     * 消息事件类
     */
    public static class MessageEvent {
        private String messageId;
        private String content;
        private String type;
        private LocalDateTime timestamp;
        private int sequenceNumber;

        // 构造器、getter、setter
        public MessageEvent() {}

        public MessageEvent(String messageId, String content, String type,
                            LocalDateTime timestamp, int sequenceNumber) {
            this.messageId = messageId;
            this.content = content;
            this.type = type;
            this.timestamp = timestamp;
            this.sequenceNumber = sequenceNumber;
        }

        // Getters and Setters
        public String getMessageId() { return messageId; }
        public void setMessageId(String messageId) { this.messageId = messageId; }

        public String getContent() { return content; }
        public void setContent(String content) { this.content = content; }

        public String getType() { return type; }
        public void setType(String type) { this.type = type; }

        public LocalDateTime getTimestamp() { return timestamp; }
        public void setTimestamp(LocalDateTime timestamp) { this.timestamp = timestamp; }

        public int getSequenceNumber() { return sequenceNumber; }
        public void setSequenceNumber(int sequenceNumber) { this.sequenceNumber = sequenceNumber; }

        @Override
        public String toString() {
            return "MessageEvent{" +
                    "messageId='" + messageId + '\'' +
                    ", content='" + content + '\'' +
                    ", type='" + type + '\'' +
                    ", timestamp=" + timestamp +
                    ", sequenceNumber=" + sequenceNumber +
                    '}';
        }
    }
}
